package com.fwmagic.flinkcdc.mysql;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @description:
 * @author: fangwei
 * @date: 2022/5/18
 **/
public class FlinkSQLCDC {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建 Flink-MYSQL-CDC
        tableEnv.executeSql("CREATE TABLE f_user (\n" +
                "  id INT primary key,\n" +
                "  name STRING,\n" +
                "  sex STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = 'hadoop3',\n" +
                "  'port' = '3306',\n" +
                "  'username' = 'root',\n" +
                "  'password' = 'qwert123',\n" +
                "  'database-name' = 'test',\n" +
                //"  'scan.startup.mode' = 'latest-offset',\n" +
                "  'table-name' = 't_user'\n" +
                ")");

        Table table = tableEnv.sqlQuery("select * from f_user");

        DataStream<Tuple2<Boolean, Row>> retractStream = tableEnv.toRetractStream(table, Row.class);

        retractStream.print();


        env.execute("FlinkSQL-CDC mysql binlog");

    }
}
